package com.xiongmm.stream.database;

import com.xiongmm.stream.support.FlinkConstant;
import com.xiongmm.stream.utils.PropertiesUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.List;

/**
 * @author XiongMM
 * @Date: 2018/6/28 16:47
 * @Description:简单的封装一下吧
 */

public final class OzoMonitorJdbc {

    protected static final Logger LOGGER = LoggerFactory.getLogger(OzoMonitorJdbc.class);

    private static String driverClassName;
    private static String URL;
    private static String username;
    private static String password;
    private static boolean autoCommit;

    /**
     * 声明一个 Connection类型的静态属性，用来缓存一个已经存在的连接对象
     */
    private static Connection conn;

    static {
        String driverClassName = PropertiesUtil.getValue("driver_class_name");
        String url = PropertiesUtil.getValue("url");
        String userName = PropertiesUtil.getValue("user_name");
        String password = PropertiesUtil.getValue("password");
        config(driverClassName, url, userName, password);
    }

    /**
     * 开头配置自己的数据库信息
     *
     * @param driverClassName
     * @param url
     * @param userName
     * @param password
     */
    private static void config(String driverClassName, String url, String userName, String password) {
        /**
         * 获取驱动
         */
        OzoMonitorJdbc.driverClassName = driverClassName;
        /**
         * 获取URL
         */
        OzoMonitorJdbc.URL = url;
        /**
         * 获取用户名
         */
        OzoMonitorJdbc.username = userName;
        /**
         * 获取密码
         */
        OzoMonitorJdbc.password = password;
        /**
         * 设置是否自动提交，一般为false不用改
         */
        OzoMonitorJdbc.autoCommit = false;
    }

    /**
     * 载入数据库驱动类
     */
    private static void load() {
        try {
            Class.forName(driverClassName);
        } catch (ClassNotFoundException e) {
            LOGGER.error("驱动类 " + driverClassName + " 加载失败");
        }
    }

    /**
     * 专门检查缓存的连接是否不可以被使用 ，不可以被使用的话，就返回 true
     */
    private static boolean invalid() {
        if (conn != null) {
            try {
                if (conn.isClosed() || !conn.isValid(FlinkConstant.INT_3)) {
                    //isValid方法是判断Connection是否有效,如果连接尚未关闭并且仍然有效，则返回 true
                    return true;
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
            //conn 既不是 null 且也没有关闭 ，且 isValid 返回 true，说明是可以使用的 ( 返回 false )
            return false;
        } else {
            return true;
        }
    }

    /**
     * 建立数据库连接
     */
    public static Connection connect() {
        //invalid为true时，说明连接是失败的
        if (invalid()) {
            try {
                //加载驱动
                load();
                //建立连接
                conn = DriverManager.getConnection(URL, username, password);
            } catch (SQLException e) {
                LOGGER.error("建立数据库连接失败 , " + e.getMessage());
            }
        }
        return conn;
    }

    /**
     * 设置是否自动提交事务
     **/
    private static void transaction() {
        try {
            conn.setAutoCommit(autoCommit);
        } catch (SQLException e) {
            LOGGER.error("设置事务的提交方式为 : " + (autoCommit ? "自动提交" : "手动提交") + " 时失败: " + e.getMessage());
        }
    }

    /**
     * 创建 Statement 对象
     */
    private static Statement statement() {
        Statement st = null;
        //如果连接是无效的就重新连接
        connect();
        //设置事务的提交方式
        transaction();
        try {
            st = conn.createStatement();
        } catch (SQLException e) {
            LOGGER.error("创建 Statement 对象失败: " + e.getMessage());
        }
        return st;
    }

    /**
     * 根据给定的带参数占位符的SQL语句，创建 PreparedStatement 对象
     *
     * @param sql 带参数占位符的SQL语句
     * @return 返回相应的 PreparedStatement 对象
     */
    private static PreparedStatement prepare(String sql, boolean autoGeneratedKeys) {
        PreparedStatement ps = null;
        //如果连接是无效的就重新连接
        connect();
        //设置事务的提交方式
        transaction();
        try {
            if (autoGeneratedKeys) {
                ps = conn.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS);
            } else {
                ps = conn.prepareStatement(sql);
            }
        } catch (SQLException e) {
            LOGGER.error("创建 PreparedStatement 对象失败: " + e.getMessage());
        }
        return ps;
    }


    public static ResultSet query(String sql, Object... params) {
        if (sql == null || sql.trim().isEmpty() || !sql.trim().toLowerCase().startsWith(FlinkConstant.JDBC_SELECT)) {
            throw new RuntimeException("你的SQL语句为空或不是查询语句");
        }
        ResultSet rs = null;
        //说明 有参数 传入，就需要处理参数
        if (params.length > 0) {
            PreparedStatement ps = prepare(sql, false);
            try {
                for (int i = 0; i < params.length; i++) {
                    ps.setObject(i + 1, params[i]);
                }
                rs = ps.executeQuery();
            } catch (SQLException e) {
                LOGGER.error("执行SQL失败: " + e.getMessage());
            }
        } else {
            //说明没有传入任何参数
            Statement st = statement();
            try {
                // 直接执行不带参数的 SQL 语句
                rs = st.executeQuery(sql);
            } catch (SQLException e) {
                LOGGER.error("执行SQL失败: " + e.getMessage());
            }
        }
        return rs;
    }


    private static Object typeof(Object o) {
        if (o == null) {
            return null;
        }
        Object r = o;
        if (o instanceof java.sql.Timestamp) {
            return r;
        }
        // 将 java.util.Date 转成 java.sql.Date
        if (o instanceof java.util.Date) {
            java.util.Date d = (java.util.Date) o;
            r = new java.sql.Timestamp(d.getTime());
            return r;
        }
        if (o instanceof java.sql.Date) {
            java.sql.Date d = (java.sql.Date) o;
            r = new java.sql.Timestamp(d.getTime());
            return r;
        }
        // 将 Character 或 char 变成 String
        if (o instanceof Character || o.getClass() == char.class) {
            r = String.valueOf(o);
            return r;
        }
        return r;
    }


    public static boolean execute(String sql, Object... params) {
        if (sql == null || sql.trim().isEmpty() || sql.trim().toLowerCase().startsWith(FlinkConstant.JDBC_SELECT)) {
            throw new RuntimeException("你的SQL语句为空或有错");
        }
        //表示 执行 DDL 或 DML 操作是否成功的一个标识变量
        boolean r = false;
        //获得 被执行的 SQL 语句的 前缀
        sql = sql.trim();
        sql = sql.toLowerCase();
        String prefix = sql.substring(0, sql.indexOf(" "));
        // 用来保存操作类型的 变量
        String operation = "";
        // 根据前缀 确定操作
        switch (prefix) {
            case "create":
                operation = "create table";
                break;
            case "alter":
                operation = "update table";
                break;
            case "drop":
                operation = "drop table";
                break;
            case "truncate":
                operation = "truncate table";
                break;
            case "insert":
                operation = "insert :";
                break;
            case "update":
                operation = "update :";
                break;
            case "delete":
                operation = "delete :";
                break;
            default:
                break;
        }
        // 说明有参数
        if (params.length > 0) {
            PreparedStatement ps = prepare(sql, false);
            Connection c = null;
            try {
                c = ps.getConnection();
            } catch (SQLException e) {
                e.printStackTrace();
            }
            try {
                for (int i = 0; i < params.length; i++) {
                    Object p = params[i];
                    p = typeof(p);
                    ps.setObject(i + 1, p);
                }
                ps.executeUpdate();
                commit(c);
                r = true;
            } catch (SQLException e) {
                LOGGER.error(operation + " 失败: " + e.getMessage());
                rollback(c);
            }
        } else {
            // 说明没有参数
            Statement st = statement();
            Connection c = null;
            try {
                c = st.getConnection();
            } catch (SQLException e) {
                e.printStackTrace();
            }
            // 执行 DDL 或 DML 语句，并返回执行结果
            try {
                st.executeUpdate(sql);
                // 提交事务
                commit(c);
                r = true;
            } catch (SQLException e) {
                LOGGER.error(operation + " 失败: " + e.getMessage());
                // 回滚事务
                rollback(c);
            }
        }
        return r;
    }

    /**
     * @param sql               需要执行的 INSERT 语句
     * @param autoGeneratedKeys 指示是否需要返回由数据库产生的键
     * @param params            将要执行的SQL语句中包含的参数占位符的 参数值
     * @return 如果指定 autoGeneratedKeys 为 true 则返回由数据库产生的键； 如果指定 autoGeneratedKeys
     * 为 false 则返回受当前SQL影响的记录数目
     */
    public static int insert(String sql, boolean autoGeneratedKeys, Object... params) {
        int var = -1;
        if (sql == null || sql.trim().isEmpty()) {
            throw new RuntimeException("你没有指定SQL语句，请检查是否指定了需要执行的SQL语句");
        }
        // 如果不是 insert 开头开头的语句
        if (!sql.trim().toLowerCase().startsWith(FlinkConstant.JDBC_INSERT)) {
            LOGGER.error(sql.toLowerCase());
            throw new RuntimeException("你指定的SQL语句不是插入语句，请检查你的SQL语句");
        }
        // 获得 被执行的 SQL 语句的 前缀 ( 第一个单词 )
        sql = sql.trim();
        sql = sql.toLowerCase();
        // 说明有参数
        if (params.length > 0) {
            PreparedStatement ps = prepare(sql, autoGeneratedKeys);
            Connection c = null;
            try {
                // 从 PreparedStatement 对象中获得 它对应的连接对象
                c = ps.getConnection();
            } catch (SQLException e) {
                e.printStackTrace();
            }
            try {
                for (int i = 0; i < params.length; i++) {
                    Object p = params[i];
                    p = typeof(p);
                    if (p == null) {
                        ps.setObject(i + 1, null);
                    } else {
                        ps.setObject(i + 1, p);
                    }
                }
                int count = ps.executeUpdate();
                // 如果希望获得数据库产生的键
                if (autoGeneratedKeys) {
                    // 获得数据库产生的键集
                    ResultSet rs = ps.getGeneratedKeys();
                    // 因为是保存的是单条记录，因此至多返回一个键
                    if (rs.next()) {
                        // 获得值并赋值给 var 变量
                        var = rs.getInt(1);
                    }
                } else {
                    // 如果不需要获得，则将受SQL影像的记录数赋值给 var 变量
                    var = count;
                }
                commit(c);
            } catch (SQLException e) {
                LOGGER.error("数据保存失败: " + e.getMessage());
                rollback(c);
            }
        } else {
            // 说明没有参数
            Statement st = statement();
            Connection c = null;
            try {
                // 从 Statement 对象中获得 它对应的连接对象
                c = st.getConnection();
            } catch (SQLException e) {
                e.printStackTrace();
            }
            // 执行 DDL 或 DML 语句，并返回执行结果
            try {
                int count = st.executeUpdate(sql);
                // 如果企望获得数据库产生的键
                if (autoGeneratedKeys) {
                    // 获得数据库产生的键集
                    ResultSet rs = st.getGeneratedKeys();
                    // 因为是保存的是单条记录，因此至多返回一个键
                    if (rs.next()) {
                        // 获得值并赋值给 var 变量
                        var = rs.getInt(1);
                    }
                } else {
                    // 如果不需要获得，则将受SQL影像的记录数赋值给 var 变量
                    var = count;
                }
                // 提交事务
                commit(c);
            } catch (SQLException e) {
                LOGGER.error("数据保存失败: " + e.getMessage());
                // 回滚事务
                rollback(c);
            }
        }
        return var;
    }

    /**
     * @param sql
     * @param autoGeneratedKeys
     * @param params
     * @return
     */
    public static boolean batchInsert(String sql, boolean autoGeneratedKeys, List<List<Object>> params) {
        if (sql == null || sql.trim().isEmpty()) {
            throw new RuntimeException("你没有指定SQL语句，请检查是否指定了需要执行的SQL语句");
        }
        // 获得 被执行的 SQL 语句的 前缀 ( 第一个单词 )
        sql = sql.trim();
        sql = sql.toLowerCase();
        PreparedStatement ps = prepare(sql, autoGeneratedKeys);
        Connection c = null;
        try {
            // 从 PreparedStatement 对象中获得 它对应的连接对象
            c = ps.getConnection();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        try {
            for (List<Object> outsideParam : params) {
                for (int i = 0; i < outsideParam.size(); i++) {
                    Object p = outsideParam.get(i);
                    p = typeof(p);
                    if (p == null) {
                        ps.setObject(i + 1, null);
                    } else {
                        ps.setObject(i + 1, p);
                    }
                }
                ps.addBatch();
            }
            ps.executeBatch();
            commit(c);
        } catch (SQLException e) {
            LOGGER.error("数据保存失败: " + e.getMessage());
            rollback(c);
            return false;
        }
        return true;
    }

    /**
     * 提交事务
     */
    private static void commit(Connection c) {
        if (c != null && !autoCommit) {
            try {
                c.commit();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 回滚事务
     */
    private static void rollback(Connection c) {
        if (c != null && !autoCommit) {
            try {
                c.rollback();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 释放资源
     **/
    public static void release(Object closeable) {
        if (closeable != null) {
            if (closeable instanceof ResultSet) {
                ResultSet rs = (ResultSet) closeable;
                try {
                    rs.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (closeable instanceof Statement) {
                Statement st = (Statement) closeable;
                try {
                    st.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (closeable instanceof Connection) {
                Connection c = (Connection) closeable;
                try {
                    c.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
